package com.anytrek.demomqtt.config;

import lombok.extern.log4j.Log4j2;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@Log4j2
@Configuration
public class MqttConfig {

    private final MqttProperties mqttProperties;

    public MqttConfig(MqttProperties mqttProperties) {
        this.mqttProperties = mqttProperties;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.getConnectionOptions().setServerURIs(new String[]{mqttProperties.getUrl()});
        factory.getConnectionOptions().setUserName(mqttProperties.getUsername());
        factory.getConnectionOptions().setPassword(
                mqttProperties.getPassword() != null ? mqttProperties.getPassword().toCharArray() : null);
        factory.getConnectionOptions().setKeepAliveInterval(mqttProperties.getKeepAliveInterval());
        factory.getConnectionOptions().setAutomaticReconnect(mqttProperties.getAutomaticReconnect());
        factory.getConnectionOptions().setCleanSession(mqttProperties.getCleanSession());
        factory.getConnectionOptions().setConnectionTimeout(mqttProperties.getConnectionTimeout());
        factory.getConnectionOptions().setMaxInflight(mqttProperties.getMaxInflight());
        
        return factory;
    }

    @Bean
    public MessageChannel mqttInboundChannel() {
        return new DirectChannel();
    }

    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttInboundAdapter(MqttPahoClientFactory mqttClientFactory) {
        String clientId = mqttProperties.getClientId() + "-inbound";
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperties.getUrl(),
                clientId,
                mqttClientFactory,
                mqttProperties.getSubTopic().split(",")
        );
        adapter.setQos(mqttProperties.getQos());
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(false); // 让 MQTT payload 以 String 形式传递
        adapter.setConverter(converter);
        adapter.setOutputChannel(mqttInboundChannel());
        adapter.setRecoveryInterval(mqttProperties.getRecoveryInterval());
        adapter.setCompletionTimeout(mqttProperties.getCompletionTimeout());
        log.info("MQTT 入站适配器初始化完成, clientId: {}, topics: {}", clientId, mqttProperties.getSubTopic());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInboundChannel")
    public MessageHandler mqttInboundHandler(ReceiverMessageHandler receiverMessageHandler) {
        return receiverMessageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MqttPahoMessageHandler mqttOutboundHandler(MqttPahoClientFactory mqttClientFactory) {
        String clientId = mqttProperties.getClientId() + "-outbound";
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(
                mqttProperties.getUrl(),
                clientId,
                mqttClientFactory
        );
        handler.setDefaultQos(mqttProperties.getQos());
        handler.setDefaultTopic(mqttProperties.getSubTopic());
        handler.setAsync(true);
        log.info("MQTT 出站处理器初始化完成, clientId: {}", clientId);
        return handler;
    }
}